package com.bw.gmall.realtime.app.ods;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONArray;
import com.alibaba.fastjson.JSONObject;
import com.bw.gmall.realtime.utils.DateFormatUtil;
import com.bw.gmall.realtime.utils.MyKafkaUtil;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.runtime.state.hashmap.HashMapStateBackend;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
import org.apache.flink.streaming.api.functions.ProcessFunction;
import org.apache.flink.util.Collector;
import org.apache.flink.util.OutputTag;

public class BaseLogAppShopOds {
    public static void main(String[] args) throws Exception {
        // TODO: 创建执行环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        // TODO: 2025/09/11 设置并行度
        env.setParallelism(1);

        // TODO: 2025/09/11 状态后端
//        env.setStateBackend(new HashMapStateBackend());
//        env.enableCheckpointing(5000);

        // TODO: 2025/09/11 从kafka读取主流数据
        String topic = "topic_log";
        String groupId = "base_log_app_group";
        DataStreamSource<String> source = env.addSource(MyKafkaUtil.getFlinkKafkaConsumer(topic, groupId));
        source.print();

        // TODO: 2025/09/12 数据清洗转换结构 定义一个“侧输出流”的标签（tag）,用于标识‘脏数据’
        OutputTag<String> dirtyStreamTag = new OutputTag<String>("dirtyStream") {
        };

        SingleOutputStreamOperator<String> filterStream = source.process(
                new ProcessFunction<String, String>() {
                    @Override
                    public void processElement(String value, ProcessFunction<String, String>.Context ctx, Collector<String> out) throws Exception {
                        try {
                            // TODO: 2025/09/12 正常情况下，直接通过out.collect()输出到主流；
                            out.collect(value);
                        } catch (Exception e) {
                            // TODO: 2025/09/12 如果发生异常（如格式不对）脏数据，通过侧输出流输出到侧输出流
                            ctx.output(dirtyStreamTag, value);
                        }
                    }
                }
        );
//        filterStream.print();

        //TODO: 2025/09/12 把字符串转换成jsonObject
        SingleOutputStreamOperator<JSONObject> mappedStream = filterStream.map(JSON::parseObject);

        // TODO: 2025/09/12 按照mid字段进行分组 分组后所属于同一个mid的记录会被发往同一个算子实例处理
        KeyedStream<JSONObject, String> keyedStream = mappedStream.keyBy(r -> r.getJSONObject("common").getString("mid"));

        SingleOutputStreamOperator<JSONObject> fixedStream = keyedStream.process(new KeyedProcessFunction<String, JSONObject, JSONObject>() {
            // TODO: 2025/9/12 每个 key即每个 mid 都有一个独立的状态变量 firstValueState，保存的是第一次访问的日期字符串
            ValueState<String> firstValueState;

            @Override
            public void open(Configuration parameters) throws Exception {
                super.open(parameters);
                firstValueState = getRuntimeContext().getState(new ValueStateDescriptor<String>("firstValueState", String.class));
            }

            @Override
            public void processElement(JSONObject value, KeyedProcessFunction<String, JSONObject, JSONObject>.Context ctx, Collector<JSONObject> out) throws Exception {
                String isNew = value.getJSONObject("common").getString("is_new");

                // TODO: 2025/9/12 获取当前状态：首次访问日期
                String firstViewDt = firstValueState.value();

                Long ts = value.getLong("ts");
                String dt = DateFormatUtil.toDate(ts);

                // TODO: 2025/9/12 认为是新用户 状态中还没有记录首次访问时间
                if ("1".equals(isNew)) {
                    if (firstViewDt == null) {
                        firstValueState.update(dt);
                    } else {
                        if (!firstValueState.equals(dt)) {
                            isNew = "0";
                            value.getJSONObject("common").put("is_new", isNew);
                        }
                    }
                } else {
                    if (firstViewDt == null) {
                        String yesterday = DateFormatUtil.toDate(ts - 1000 * 60 * 60 * 24);
                        firstValueState.update(yesterday);
                    }
                }
                out.collect(value);
            }
        });

        //Todo: 2025/9/12 分流 page start err action display
        OutputTag<String> startTag = new OutputTag<String>("startTag") {
        };
        OutputTag<String> errTag = new OutputTag<String>("errTag") {
        };
        OutputTag<String> actionTag = new OutputTag<String>("actionTag") {
        };
        OutputTag<String> displayTag = new OutputTag<String>("displayTag") {
        };
        
//        fixedStream.process(new ProcessFunction<JSONObject, String>() {
//            @Override
//            public void processElement(JSONObject value, ProcessFunction<JSONObject, String>.Context ctx, Collector<String> out) throws Exception {
//                // TODO: 2025/9/12 提取err字段：如果存在，则说明是一条错误日志
//                JSONObject err = value.getJSONObject("err");
//                if (err!=null){
//                    ctx.output(errTag,value.toJSONString());
//                }
//                value.remove("err");
//
//                // TODO: 2025/9/12 提取start字段：如果存在，则说明是启动日志
//                JSONObject start = value.getJSONObject("start");
//                if (start!=null){
//                    ctx.output(startTag,value.toJSONString());
//                }
//                else {
//                    // TODO: 2025/9/12 提取公共字段
//                    JSONObject page = value.getJSONObject("page");
//                    JSONObject common = value.getJSONObject("common");
//                    Long ts = value.getLong("ts");
//
//                    // TODO: 2025/9/12 收集曝光数据[{},{},{}]display是一个数组，每一个元素是一个曝光事件（比如广告，商品展示）
//                    JSONArray displays = value.getJSONArray("displays");
//                    if (displays!=null){
//                        // TODO: 2025/9/12 遍历每一个曝光事件，构建新的JSON对象
//
//                    }
//                }
//
//            }
//        });


        env.execute();

    }

}
























